package com.yasaka.stock.kafkaConsumer;

import java.util.concurrent.Callable;
import java.util.concurrent.Semaphore;

public class WorkThread implements Callable<String> {
    /**
     * 记录解析失败的日志或者发送到kafka集群失败的日志
     */
    private String topic;
    private String message;
    private Semaphore semaphore;

    public WorkThread(String topic, String message, Semaphore semaphore) {
        this.topic = topic;
        this.message = message;
        this.semaphore = semaphore;
    }

    @Override
    public String call() throws Exception {
        try {
            //deal your business
            //......

            //这里只打印下，如果在实际业务中处理失败，可能是代码bug或者系统不稳定等，
            //先将消息记录到日志中，后续可以处理,不影响主流程的继续运行
            System.out.println("topic is {" + topic + "}, message is {" + message + "}");
        } catch (Exception e) {
            System.out.println("ParseKafkaLogJob run error. " + e);
        } finally {
            semaphore.release();
        }
        return "done";
    }
}